package com.qupeng.demo.kafka.kafkaspringbootstreams.common.kafka.streams;

import org.apache.kafka.streams.processor.Processor;
import org.apache.kafka.streams.processor.ProcessorContext;
import org.apache.kafka.streams.processor.PunctuationType;
import org.apache.kafka.streams.processor.Punctuator;
import org.apache.kafka.streams.processor.To;

import java.time.Duration;

public class MyProcessor implements Processor<String, String> {
    private ProcessorContext context;

    @Override
    public void init(ProcessorContext context) {
        this.context = context;

        // 定时触发 punctuate 方法，以便处理时间戳
        this.context.schedule(Duration.ofSeconds(1), PunctuationType.WALL_CLOCK_TIME, new Punctuator() {
            @Override
            public void punctuate(long timestamp) {
                // 在这里处理时间戳
//                System.out.println("Current event timestamp: " + context.timestamp());
            }
        });
    }

    @Override
    public void process(String key, String value) {
        // 处理输入记录
        System.out.println("Received record with timestamp: " + context.timestamp());
    }

    @Override
    public void close() {
        // 关闭处理器
    }
}
